
package com.ibm.cps.messagequeue;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.common.base.Throwables;
import com.ibm.interfaces.IMessageClient;
import com.ibm.util.ErrorCode;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

public class AMQPClient implements IMessageClient, KryoSerializable{

	private ConnectionFactory factory;
	private Connection connection;
	private Channel channel;
	private Vector<String> existedQueues;
	private Logger logger;
	private QueueingConsumer consumer;
	
	private String server;
	private int port;
	private String clientName;
	private String password;

	public AMQPClient() throws CPSException {
		logger = Logger.getLogger(AMQPClient.class);
		PropertyConfigurator.configure(LocalConfig.getLoggerProperties());
		logger.info("Using amqp client.");
	}

	public AMQPClient(String server, int port, String clientName, String password) throws CPSException {
		this.server = server;
		this.port = port;
		this.clientName = clientName;
		this.password = password;
		this.existedQueues = new Vector<String>();
		createConnections();
	}
	
	private void createConnections() throws CPSException{
		PropertyConfigurator.configure(LocalConfig.getLoggerProperties());
		logger = Logger.getLogger(AMQPClient.class);
		factory = new ConnectionFactory();
		factory.setHost(server);
		factory.setPort(port);
		factory.setUsername(clientName);
		factory.setPassword(password);
		try {
			connection = factory.newConnection();
			channel = connection.createChannel();
			consumer = new QueueingConsumer(channel);
			logger.info("Connected to " + server + ":" + port + " with client ID " + clientName);
			for(String topic : existedQueues){
				subscribe(topic);
			}
		} catch (IOException e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Can't get connection to server " + server + ":" + port + " successfully.");
		}
	}
	
	@Override
	public void read(Kryo kryo, Input input) {
		this.server = input.readString();
		this.port = input.readInt();
		this.clientName = input.readString();
		this.password = input.readString();
		this.existedQueues = new Vector<String>();
		int topicCount = input.readInt();
		for (int i = 0; i < topicCount; i++) {
			existedQueues.add(input.readString());
		}
		try {
			createConnections();
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(server);
		output.writeInt(port);
		output.writeString(clientName);
		output.writeString(password);
		output.writeInt(existedQueues.size());
		for(String topicName : existedQueues){
			output.writeString(topicName);
		}
	}

	@Override
	public void publishMessage(String topicName, byte[] message) throws CPSException {
		if (channel == null) {
			throw new CPSException(ErrorCode.NULL_CHANNEL, "No valid channel.");
		}
		try {
			if (!existedQueues.contains(topicName)) {
				channel.exchangeDeclare(topicName, "fanout", false, false, null);
				channel.queueDeclare(topicName, false, false, false, null);
				channel.queueBind(topicName, topicName, topicName);
				existedQueues.add(topicName);
			}
			channel.basicPublish(topicName, topicName, null, message);
			logger.info("Send message " + new String(message) + " to topic " + topicName);
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	@Override
	public void subscribe(String topicName) throws CPSException {
		try {
			checkChannel();
			if (existedQueues.contains(topicName)) {
				logger.info(topicName + " already been added.");
				return;
			}
			// channel.exchangeDeclare(topicName, "fanout");
			// channel.exchangeDeclare(topicName, "fanout", false, true, null);
			// channel.queueDeclare(topicName, false, false, true, null);
			channel.exchangeDeclare(topicName, "fanout", false, false, null);
			long time = System.currentTimeMillis();
			String queueName = topicName + time;
			channel.queueDeclare(queueName, false, false, false, null);
			channel.queueBind(queueName, topicName, topicName);
			existedQueues.add(queueName);
			String returnValue = channel.basicConsume(queueName, false, queueName, consumer);
			logger.info("Topic " + topicName + " has been added.");
			logger.info("Consumer tag is " + consumer.getConsumerTag());
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.MESSAGE_QUEUE_ERROR, e.getMessage());
		}
	}

	private void checkChannel() throws CPSException {
		if (channel == null) {
			throw new CPSException(ErrorCode.SERVER_NOT_FOUND_CODE, "No valid channel.");
		}
	}

	@Override
	public void unsubscribe(String topic) throws CPSException {
		try {
			if (!existedQueues.contains(topic)) {
				logger.error(topic + " doesn't exist.");
				return;
			}
			existedQueues.remove(topic);
			channel.basicCancel(topic);
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.MESSAGE_QUEUE_ERROR, "Can't remove queue " + topic + ".");
		}
	}

	@Override
	public Collection<String> consumeMessage() throws CPSException {
		try {
			ArrayList<String> result = new ArrayList<String>();
			Delivery delivery = consumer.nextDelivery(10000);
			int threshold = 10000;
			long start = System.currentTimeMillis();
			long time = 0;
			while(time<threshold)
			{
				if (delivery == null) {
					return null;
				}
				// Delivery delivery = consumer.nextDelivery();
				byte[] message = delivery.getBody();
				String topic = delivery.getEnvelope().getExchange();
				if (topic == null || topic.length() < 1) {
					topic = delivery.getEnvelope().getRoutingKey();
				}
				if (topic == null || topic.length() < 1) {
					return null;
				}
				result.add(new String(message));
				delivery = consumer.nextDelivery(10000);
				long end = System.currentTimeMillis();
				time = end - start;
			}
			//null means receiving timeout
			if(delivery != null){
				channel.basicAck(delivery.getEnvelope().getDeliveryTag(), true);
			}
			return result;
		} catch (Exception e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Internal error of rabbitmq.");
		}
	}

	@Override
	public void publishMessage(String destination, String message) throws CPSException {
		publishMessage(destination, message.getBytes());
	}

	@Override
	public void close() {
		try {
			if(consumer.getConsumerTag() != null) {
				channel.basicCancel(consumer.getConsumerTag());
			}
			// Old code that seems to close multiple connection to a same topic
//			for (String ds : existedQueues) {
//
//			}
			channel.close();
			connection.close();
		} catch (IOException e) {
			logger.error(Throwables.getStackTraceAsString(e));
			logger.info("Connection was closed.");
		}
	}

}
